package cn.lgwen.candy.runtime.core;

import cn.lgwen.candy.runtime.operator.Processor;
import cn.lgwen.candy.runtime.operator.SinkOperator;

import java.util.LinkedList;
import java.util.List;

/**
 * 2019/12/24
 * aven.wu
 * danxieai258@163.com
 * description a stream
 * stream 是抽象，表示表示一个处理
 * T 表示输入类型
 */
public class Stream<T> {

    protected Environment environment;

    protected Processor<?, ?> processor;

    protected SinkOperator<T> sinkOperator;
    // 下游
    protected List<Stream<?>> nextStreams;

    protected int parallel = 1;

    protected String name;

    void parallel(int parallel) {
        if (parallel < 1) {
            throw new IllegalArgumentException("parallel must greater then 0");
        }
        this.parallel = parallel;
    }

    public int getParallel() {
        return parallel;
    }

    public Stream(Environment environment) {
        nextStreams = new LinkedList<>();
        this.environment = environment;
    }

    /**
     * 输入一个 operator 输入类型是T 输出对应 out类型的 stream
     * @param operator
     * @param
     */
    public <R> Stream<R> processor(Processor<T, R> operator) {
        Stream<R> nextStream = new Stream<>(this.environment);
        nextStream.processor = operator;
        nextStreams.add(nextStream);
        return nextStream;
    }

    public void addSink(SinkOperator<T> sinkOperator) {
        this.sinkOperator = sinkOperator;
    }

    public List<Stream<?>> getNextStreams() {
        return nextStreams;
    }

    public String getName() {
        return name;
    }

    public void name(String name) {
        this.name = name;
    }
}
